package com.lsh.payment.worker.kafka.producer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * Created by wuhao on 2019/10/18.
 */
@Service
public class ProducerService{
    private static final Logger logger = LoggerFactory.getLogger(ProducerService.class);

    private Gson gson = new GsonBuilder().create();

    @Autowired
    private KafkaTemplate template;

    @Value("${topic_name}")
    public String topic;

    //发送消息方法
    public void sendJson(String json) {
        JSONObject jsonObj = JSON.parseObject(json);

        jsonObj.put("topic", topic);
        jsonObj.put("ts", System.currentTimeMillis() + "");

        logger.info("send kafka message = {}", jsonObj.toJSONString());

        ListenableFuture<SendResult<String, String>> future = template.send(topic, jsonObj.toJSONString());
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                logger.info("send kafka message success result:{}", result);
            }

            @Override
            public void onFailure(Throwable ex) {
                logger.info("send kafka message fail , error msg: {}", ex.getMessage());
            }
        });
    }

//    @Override
//    public void afterPropertiesSet() throws Exception {
//        Map<String,Object> params = new HashMap<>();
//        params.put("123", 123);
//        for(int i=0;i<100;i++) {
//            this.sendJson(JSON.toJSONString(params));
//        }
//    }
}
